package sos.entrega2013.mr1.noviembre.bo.toptres;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import sos.entrega2013.mr1.noviembre.beans.CDR;

public class TopTresReduce extends MapReduceBase implements Reducer<Text, CDR, Text, Text>{
	 
	@Override
	public void reduce(Text key, Iterator<CDR> values,	OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
		HashMap<String, Integer> hmMap=new HashMap<String, Integer>();
		int iDuracionLlamada=0;
		String [][] ordenado;
		while(values.hasNext()){
			CDR llamada=values.next();
			String numero=null;
			if(llamada.getLlamante().equals(key.toString())){
				numero=llamada.getLlamado();
			}
			else if(llamada.getLlamado().equals(key.toString())){
				numero=llamada.getLlamante();
			}
			if(hmMap.containsKey(numero)){
				iDuracionLlamada=Integer.valueOf(hmMap.get(numero).toString())+llamada.getDuracion();
				hmMap.put(numero, iDuracionLlamada);
			}
			else{
				hmMap.put(numero, llamada.getDuracion());
			}	
		}
		ordenado=this.ordenacion(hmMap);
		for(int i=0; i<ordenado.length; i++){
			if(i<3){
				output.collect(key, new Text(ordenado[i][0]+","+ordenado[i][1]));
			}
			else{
				break;
			}
		}
		
	}
	public String[][] ordenacion(HashMap<String,Integer> map){
		int i=0, valor;
		String key;
		String [][] dataMap=new String[map.size()][2];
		Iterator it = map.entrySet().iterator();
		//Pasamos todos los datos del map a un array
		while(it.hasNext()){
			Entry<String,Integer> e=(Entry<String,Integer>)it.next();
			dataMap[i][0]=e.getKey();
			dataMap[i][1]=e.getValue().toString();
			i++;
		}
		//ordenamos el array
		for(int m=0; m<dataMap.length;m++){
			for(int k=0;k<m;k++){
				if(Integer.valueOf(dataMap[m][1])>Integer.valueOf(dataMap[k][1])){
					key 	=	dataMap[k][0];
					valor	=	Integer.valueOf(dataMap[k][1]);
					dataMap[k][0]=dataMap[m][0];
					dataMap[k][1]=dataMap[m][1];
					dataMap[m][0]=key;
					dataMap[m][1]=String.valueOf(valor);
				}
			}
			
		}
		return dataMap;
	}

}
